package com.cl.ks.schedule;

import com.cl.ks.entity.KsFlowTask;
import com.cl.ks.repository.KsFlowTaskRepository;
import com.cl.ks.service.SparkNodeService;
import com.cl.schedule.schedule.BaseStarter;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class KsFlowTaskStarter extends BaseStarter<KsFlowTask> {

    private final KsFlowTaskRepository ksFlowTaskRepository;

    private final SparkNodeService sparkNodeService;


    public KsFlowTaskStarter(KsFlowTaskRepository ksFlowTaskRepository, SparkNodeService sparkNodeService) {
        this.ksFlowTaskRepository = ksFlowTaskRepository;
        this.sparkNodeService = sparkNodeService;
    }


    public void startAll() {
        List<KsFlowTask> ksFlowTaskList = ksFlowTaskRepository.findAll();
        for (KsFlowTask ksFlowTask : ksFlowTaskList) {
            start(ksFlowTask);
        }
    }

    @Override
    public void start(KsFlowTask ksFlowTask) {
        Integer ksFlowGraphId = ksFlowTask.getKsFlowGraphId();
        String createdBy = ksFlowTask.getCreatedBy();
        String showName = ksFlowTask.getCreatedByShowName();

        if (ksFlowGraphId == null) {
            return;
        }
        sparkNodeService.handleKsFlowGraphById(ksFlowTask,ksFlowGraphId,createdBy,showName);

    }
}
